package com.xzc.apitest.checkPoint

import java.util.concurrent.TimeUnit

import com.xzc.apitest.source.SensorReading
import org.apache.flink.api.common.restartstrategy.RestartStrategies
import org.apache.flink.api.common.time.Time
import org.apache.flink.streaming.api.CheckpointingMode
import org.apache.flink.streaming.api.scala._

object CheckPointTest {
  def main(args: Array[String]): Unit = {
    //创建执行环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    //默认是MemoryStateBackend
    //    env.setStateBackend(new FsStateBackend("11"))
    //    env.setStateBackend(new RocksDBStateBackend("123"))
    //插入B的时间，头到头的时间间隔，如果超过则推移
    env.enableCheckpointing(1000L)
    //至少正确一次，对正确要求不是那么严格的时候使用
    env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.AT_LEAST_ONCE)
    //等一分钟，没完成就丢弃
    env.getCheckpointConfig.setCheckpointTimeout(60000L)
    //默认为1个，如果为1个就是要等全流程CP做完，JOBM才能触发下一个CP，而设置>1后，就相当于PIPLINE的概念了
    env.getCheckpointConfig.setMaxConcurrentCheckpoints(2)
    //尾到头的时间间隔，setMaxConcurrentCheckpoints这个就没有用了，因为只能一个一个的来
    env.getCheckpointConfig.setMinPauseBetweenCheckpoints(500L)
    //优先使用CP来恢复，默认为false，SP早于CP的话会用SP
    env.getCheckpointConfig.setPreferCheckpointForRecovery(true)
    //默认为0，一旦CP挂了，就认为JOB挂了
    env.getCheckpointConfig.setTolerableCheckpointFailureNumber(3)

    //故障重启
    //固定间隔时间重启，这里是10秒，重启3秒，
    env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 10000L))
    //5分钟内重启5次，每次重启之间的间隔为10秒
    env.setRestartStrategy(RestartStrategies.failureRateRestart(
      5, Time.of(5, TimeUnit.MINUTES), Time.of(10, TimeUnit.SECONDS)
    ))

    val inputStream = env.socketTextStream("hadoop102", 7777);

    val dataStream = inputStream
      .map(data => {
        val arr = data.split(",")
        SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
      })
  }
}
